/* 没有定义__MS_REACTOR_H__再进行编译，如果定义了，说明头文件已经包含过了，不用再进行编译了，否则就重复包含头文件了 */
#ifndef __MS_REACTOR_H__
#define __MS_REACTOR_H__
#include <iostream>
#include <algorithm>
#include <functional>
#include <string>
#include <vector>
#include <unordered_map>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <memory>
#include <utility>
#include <assert.h>
#include <boost/noncopyable.hpp>
#include <string.h>
#include <time.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <signal.h>

/* 1.我们的这个高性能主从Reactor服务器组件写完之后，如果要提供给其他人使用，则有两种方式，一种是提供头文件.h + 库文件.o，别人在使用的时候
包含.h文件，然后在链接的时候链接上动态库即可，动态库里面其实就是我们所实现的各种方法编译之后的二进制代码，也就是方法的集合。
   2.而另一种也很常见的方式就是把方法的实现和方法的声明合在一起，搞成一个.hpp文件，对方直接包含这个.hpp文件就可以使用，而无需链接额外的库，
库其实就是.o文件集合
   3.我们的组件提供方式采用第二种，这样比较简单，因为第一种还得先编译，然后再打包.o文件，而不同的打包工具和打包方式还会划分出
静态库和动态库，感觉有点麻烦，使用者使用的时候还得链接库，不是很方便，所以我们采用第二种方式来提供给使用者 */

#define NOR 0
#define DBG 1
#define ERR 2
#define LOGLEVEL DBG

#define LOG(level, format, ...)                                                                 \
    do                                                                                          \
    {                                                                                           \
        if (level >= LOGLEVEL)                                                                  \
        {                                                                                       \
            time_t t = time(NULL);                                                              \
            struct tm *local = localtime(&t);                                                   \
            char tmp[32] = {0};                                                                 \
            strftime(tmp, sizeof(tmp) - 1, "%H:%M:%S", local);                                  \
            pthread_t tid = pthread_self();                                                     \
            fprintf(stdout, "[%lu %s %s:%d] " format "\n", (unsigned long)tid, tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
        }                                                                                       \
    } while (false)

#define NLOG(format, ...) LOG(NOR, format, ##__VA_ARGS__)
#define DLOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_DEFAULT_SIZE 1024
class Buffer
{
private:
    uint64_t _read_idx;
    uint64_t _write_idx;
    std::vector<char> _buffer;
public:
    Buffer() : _read_idx(0), _write_idx(0), _buffer(BUFFER_DEFAULT_SIZE) {}
    char *WriteAddress() { return &_buffer[_write_idx]; }
    const char *ReadAddress() const { return &_buffer[_read_idx]; }
    char *ReadAddress() { return &_buffer[_read_idx]; }
    uint64_t BackRemain() { return _buffer.size() - _write_idx; }    // 获取_write_idx后面的空闲空间大小 
    uint64_t FrontRemain() { return _read_idx; }                     // 获取_read_idx前面的空闲空间大小 
    uint64_t ReadableSize()const { return _write_idx - _read_idx; }  //开区间-闭区间
    
    // 读取数据之后，将_read_idx向后移动到正确的位置
    void MoveReadIdx(uint64_t len)
    {
        assert(len <= ReadableSize());
        _read_idx += len;
    }
    // 写入数据之后，将_write_idx向后移动到正确的位置
    void MoveWriteIdx(uint64_t len)
    {
        // 向后移动的大小必须小于等于后边剩余空间大小，因为在移动写偏移之前，一定是写入数据成功了，写入数据之前一定会确保写空间足够，
        // 在确保空间足够这里，要么扩容，要么挪动数据，最终都会保证_write_idx后面的可写空间是足够的，所以这里只需要判断BackRemain即可
        assert(len <= BackRemain());
        _write_idx += len;
    }
    // 确保可写空间是足够的(1.整体空余空间真的不够了，直接向后扩容 2.整体空余空间够，将数据向前挪动)
    void EnsureWriteSpace(uint64_t size) // 外部给一个确保空间的大小，内部通过size和空余空间来作比较，分情况处理
    {
        if (BackRemain() >= size)
        {
            // 后面剩余的空间都够了，啥都不用干就行
        }
        else if (FrontRemain() + BackRemain() >= size)
        {
            // 整体空间够: 扩容+移动读写偏移
            uint64_t rsz = ReadableSize();
            // 迭代器其实就是对原生指针的封装。 注意: copy接口的第二个参数是开区间
            std::copy(ReadAddress(), ReadAddress() + rsz, &_buffer[0]);
            _read_idx = 0, _write_idx = rsz; //_write_idx是开区间
        }
        else
        {
            // 整体空间不够: 不移动数据，直接向后扩容足够空间即可
            _buffer.resize(_write_idx + size); // 异地扩容之后，容器会帮我们进行数据的拷贝，我们自己不用操心
        }
        return;
    }
    // 写入数据
    void Write(const void *data, uint64_t size)
    {
        // 1.保证有足够写入的空间
        EnsureWriteSpace(size);
        // 2.拷贝数据(地址加减整数，其实就是让指针挪动数据个数的大小，比如指针指向0idx，加上3，则指向4idx)
        std::copy((const char *)data, (const char *)data + size, WriteAddress());
    }
    void WriteAndMoveIdx(const void *data, uint64_t size) // 写入数据+挪动写偏移量，相当于写入数据时，把数据推进Buffer里面
    {
        Write(data, size);
        MoveWriteIdx(size);
    }
    void Write(const std::string &data)
    {
        Write(data.c_str(), data.size());
    }
    void WriteAndMoveIdx(const std::string &data)
    {
        Write(data);
        MoveWriteIdx(data.size());
    }
    void Write(const Buffer &data)
    {
        Write(data.ReadAddress(), data.ReadableSize());
    }
    void WriteAndMoveIdx(const Buffer &data)
    {
        Write(data);
        MoveWriteIdx(data.ReadableSize());
    }
    void Read(void *buf, uint64_t size)
    {
        assert(size <= ReadableSize());
        std::copy(ReadAddress(), ReadAddress() + size, (char *)buf);
    }
    void ReadAndMoveIdx(void *buf, uint64_t size) // 读取数据+挪动读偏移量，相当于读取数据之后，把读取到的数据从Buffer里面弹出去
    {
        Read(buf, size);
        MoveReadIdx(size);
    }
    std::string Read(uint64_t size)
    {
        assert(size <= ReadableSize());
        std::string str;
        str.resize(size);
        Read(&str[0], size);
        return str;
    }
    std::string ReadAndMoveIdx(uint64_t size) // 读取数据+挪动读偏移量，相当于读取数据之后，把读取到的数据从Buffer里面弹出去
    {
        assert(size <= ReadableSize());
        std::string str = Read(size);
        MoveReadIdx(size);
        return str;
    }
    // 把缓冲区中的内容读取出来，然后放到一个空的Buffer里面，这个Read接口我们就不实现了。
    // 因为这个功能完全可以调用空Buffer里面的Write接口，将当前Buffer中的数据写到这个空的Buffer里面，当前Buffer作为输入型参数传递到Write接口里面。
    // 清空缓冲区中的所有数据
    void Clear() { _read_idx = _write_idx = 0; }
    /* 由于后面我们的高并发服务器会支持应用层协议HTTP，而在HTTP中经典的操作就是读取一行的数据，因为请求行和请求报头以及响应行和响应报头都是
    以\r\n作为分隔符的，都是一行一行的数据，所以我们的缓冲区这里也对外提供一个接口，用于每次读取一行的数据 */
    char *FindCRLF()
    {
        void *res = memchr(ReadAddress(), '\n', ReadableSize());
        return (char *)res;
    }
    /* 通常获取一行数据，这种情况针对的是ascll码字符，所以我们这里直接返回string了，但在其他的应用层协议情况下，一行数据可能不是ascll码
    但至少在HTTP这里，我们直接返回string大概率是不会出问题的 */
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == NULL) return "";
        /* 把一行数据最后的\n字符也读取出来。指针做差，求的是两个指针间的类型数据的个数，比如int数组中，0下标和1下标的地址差确实为4字节，
        但当0下标和1下标的地址实际做差时，得到的结果会是1，而不是4字节。在下面这个地方，读取的数据的大小个数就是两个地址之间做差后得到的值
        因为指针做差刚好拿到的就是该类型数据的个数 */
        return Read(pos - ReadAddress() + 1);
    }
    std::string GetLineAndMoveIdx()
    {
        std::string str = GetLine();
        MoveReadIdx(str.size());
        return str;
    }
};

#define MAX_LISTEN 1024
class Socket
{
private:
    int _sockfd;
public:
    Socket() : _sockfd(-1) {}
    Socket(int fd) : _sockfd(fd) {}
    ~Socket() { Close(); } // 调用Close关闭套接字
    int Fd() { return _sockfd; }
    // 1.创建套接字
    bool CreateSocket()
    {
        // int socket(int domain, int type, int protocol);
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ELOG("CREATE SOCKET FAILED!!");
            return false;
        }
        return true;
    }
    // 2.绑定地址信息
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in local;
        local.sin_family = AF_INET;   // ipv4地址域类型
        local.sin_port = htons(port); // short大小，主机转网络字节序
        local.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        // int bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
        int ret = bind(_sockfd, (const struct sockaddr *)&local, len); // 权限缩小，非const到const
        if (ret < 0)
        {
            ELOG("BIND ADDRESS FAILED!!");
            return false;
        }
        return true;
    }
    // 3.开始监听连接是否到来
    bool Listen(int backlog = MAX_LISTEN)
    {
        // int listen(int sockfd, int backlog);
        // backlog是内核监听连接队列的最大长度，放着的都是完成三次握手后的连接，等待被accept拿上去进行通信
        int ret = listen(_sockfd, backlog);
        if (ret < 0)
        {
            ELOG("SOCKET LISTEN FAILED!!");
            return false;
        }
        return true;
    }
    // 4.向服务器发起连接
    bool Connect(const std::string &ip, uint16_t port) // 传进来服务器的ip和端口信息
    {
        // int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
        struct sockaddr_in peer;
        peer.sin_family = AF_INET;
        peer.sin_port = htons(port);
        peer.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_sockfd, (const struct sockaddr *)&peer, len);
        if (ret < 0)
        {
            ELOG("CONNECT SERVER FAILED!!");
            return false;
        }
        return true;
    }
    // 5.监听有连接到来后，获取新连接
    int Accept() // 返回一个文件描述符
    {
        // int accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen); 后两个参数是输出型参数，用于给服务器返回客户端的地址信息
        // 目前服务器并不关心客户端的地址信息，所以我们直接传NULL
        int newfd = accept(_sockfd, NULL, NULL);
        if (newfd < 0)
        {
            ELOG("SOCKET ACCEPT FAILED!!");
            return -1;
        }
        return newfd;
    }
    // 6.接收数据(size_t是无符号整数，ssize_t是有符号整数)
    ssize_t Recv(void *buf, size_t len, int flag = 0) // 标志位可以将接收数据设置为非阻塞操作，默认是0阻塞操作
    {
        // ssize_t recv(int sockfd, void *buf, size_t len, int flags);
        ssize_t ret = recv(_sockfd, buf, len, flag);
        if (ret <= 0)
        {
            // 只有sockfd是非阻塞的情况下进行读取，才会出现EAGAIN这种错误，当前内核缓冲区中没有数据了，希望你再读一次数据
            if (errno == EAGAIN || errno == EINTR)
                return 0; // socket内核缓冲区没有数据，或者当前系统调用被信号打断，转而执行信号处理函数去了，那么这种不算错误
            else
            {
                ELOG("SOCKET RECV FAILED!!");
                return -1;
            }
        }
        return ret; // 返回实际接收数据长度
    }
    ssize_t NonBlockRecv(void *buf, size_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT); // 接收数据操作为非阻塞
    }
    // 7.发送数据
    ssize_t Send(const void *buf, size_t len, int flag = 0) // 标志位可以将接收数据设置为非阻塞操作，默认是0阻塞操作
    {
        // ssize_t send(int sockfd, const void *buf, size_t len, int flags);
        int ret = send(_sockfd, buf, len, flag);
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return 0; // socket内核缓冲区没有空间，或者当前系统调用被信号打断，转而执行信号处理函数去了，那么这种不算错误
            else
            {
                ELOG("SOCKET SEND FAILED!!");
                return -1;                
            }
        }
        return ret; // 实际发送的数据长度，有可能实际发送的长度比想要发送的长度len要小，因为缓冲区中可能没有len那么多的数据
    }
    ssize_t NonBlockSend(void *buf, size_t len)
    {
        if(len <= 0) return 0;
        return Send(buf, len, MSG_DONTWAIT); // 发送数据操作为非阻塞，如果socket内核接收缓冲区中没有多余空间，则非阻塞返回后，errno为EAGAIN
    }
    // 8.关闭套接字
    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }
    // 9.创建服务端连接   // 服务器的ip地址默认绑定服务器主机下的所有网卡
    bool CreateServerConn(uint16_t port, bool block_flag = false, const std::string &ip = "0.0.0.0") //默认是阻塞操作
    {
        // 1.创建套接字  2.绑定地址信息  3.开始监听  4.设置套接字为非阻塞  5.设置套接字为地址重用
        if (CreateSocket() == false) return false;
        if (block_flag) SetNonBlock();
        if (Bind(ip, port) == false) return false;
        if (Listen() == false) return false;
        SetReuseAddress();
        return true;
    }
    // 10.创建客户端连接
    bool CreateClientConn(uint16_t port, const std::string &ip)
    {
        // 1.创建套接字  2.绑定地址信息(让系统自动分配端口，我们不显示的绑定)  3.发起连接
        if (CreateSocket() == false)
            return false;
        if (Connect(ip, port) == false)
            return false;
        return true;
    }
    // 11.设置套接字选项为地址重用
    void SetReuseAddress()
    {
        // int setsockopt(int sockfd, int level, int optname, const void *optval, socklen_t optlen);
        int optval = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&optval, sizeof(int));
        optval = 1;
        int ret = setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&optval, sizeof(int));
        // SO_REUSEPORT避免了应用层多线程或者进程监听同一ip:port时的"惊群效应"
    }
    // 12.设置文件描述符为非阻塞操作 (防止recv不断读取数据，使得内核缓冲区中没有数据时，执行流阻塞在recv读取这块儿)
    void SetNonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
};

class Poller;
class EventLoop;
class Channel//套接字的事件管理类
{
    using EventCallBack = std::function<void()>;
private:
    int _fd;
    EventLoop *_loop;//(connection : Channel : EventLoop : thread = 1 : 1 : 1 : 1)
    uint32_t _events;//当前需要监控的事件集合
    uint32_t _revents;//当前连接上触发的事件集合，也就是已经就绪的事件集合
    EventCallBack _read_cb;//socket上可读事件就绪后的回调
    EventCallBack _write_cb;//socket上可写事件就绪时的回调 
    EventCallBack _error_cb;//socket上错误事件就绪时的回调 
    EventCallBack _close_cb;//socket上连接关闭事件就绪时的回调 
    EventCallBack _anyevent_cb;//socket上任意事件就绪时的回调(刷新活跃度)
public:
    Channel(EventLoop *loop, int fd) 
        :_fd(fd), _events(0), _revents(0), _loop(loop)
    {}
    int Fd() { return _fd; }
    void SetRevents(uint32_t revents) { _revents = revents; }//设置实际就绪的事件
    uint32_t GetEvents(){ return _events; }//获取想要监控的事件
    void SetReadCallBack(const EventCallBack &cb) { _read_cb = cb; }
    void SetWriteCallBack(const EventCallBack &cb){ _write_cb = cb;}
    void SetErrorCallBack(const EventCallBack &cb){ _error_cb = cb;}
    void SetCloseCallBack(const EventCallBack &cb){ _close_cb = cb;}
    void SetAnyEventCallBack(const EventCallBack &cb){ _anyevent_cb = cb;}
    bool IsReadable() { return (_events & EPOLLIN); }//当前连接是否监控可读
    bool IsWritable() { return (_events & EPOLLOUT); }//当前连接是否监控可写
    void CareReadEvent() { _events |= EPOLLIN; UpdateEvents(); }//关心当前连接上的读事件
    void CareWriteEvent() { _events |= EPOLLOUT; UpdateEvents(); }//关心当前连接上的写事件
    void CancelReadEvent() { _events &= ~EPOLLIN; UpdateEvents(); }//取消读事件的关心
    void CancelWriteEvent() { _events &= ~EPOLLOUT; UpdateEvents(); }//取消写事件的关心
    void CancelAllEvents() { _events = 0; UpdateEvents(); }//取消连接上所有的关心事件
    void RemoveEvents();
    void UpdateEvents();
    
    void HandleReadyEvent()//eventloop不关心如何处理就绪的事件，Channel自己负责，EventLoop只负责调用Channel里的HandleReadyEvent()
    {
        /* 后面实现时，读取出错是不会释放连接的，即使对方挂断连接，我们读取出错，我们也要看outbuffer中是否还有待send的data，如果有,我们还要
        将待send的data进行发送，此时给已经断开连接的对方发送数据自然会出错，这个时候我们才会去断开连接。
        （所以不会出现由于重复释放连接，而导致segmentation fault） */
        if((_revents & EPOLLIN) || (_revents & EPOLLPRI) || (_revents & EPOLLRDHUP))//RDHUP -> Read Hang Up
        {
            if(_read_cb) _read_cb(); 
        }
        /* 有可能当前触发了可写事件，但写入数据不成功，对方已经挂断了连接，此时如果继续向下进行处理则会导致重复释放，所以是if else*/
        if(_revents & EPOLLOUT)
        {
            if(_write_cb) _write_cb();
        }
        else if(_revents & EPOLLERR)
        {
            if(_error_cb) _error_cb();
        }
        else if(_revents & EPOLLHUP)
        {
            if(_close_cb) _close_cb();
        }

        /* 在刷新活跃度之前，即使连接已经断开，但上面的接口都不能真正把连接释放，因为下面还会进行连接活跃度的刷新，所以在上面的回调函数中
        必须将实际释放连接的操作压入到任务池，在就绪事件全部处理完毕之后，再下一步执行任务池中的任务时，才能进行实际连接的释放 */
        if(_anyevent_cb) _anyevent_cb();//fd上任意事件触发，刷新连接的活跃度
    }
};

#define MAX_EPOLLEVENTS 1024
class Poller : boost::noncopyable//描述符fd上的IO事件监控
{
private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS];//输出型参数数组，epoll_wait会把就绪的事件结构体返回到我们传入的数组中
    std::unordered_map<int, Channel *> _chans;
    //建立fd与channel的映射关系是因为，当epoll_wait检测到部分的fd就绪之后，我们需要将就绪的事件添加到fd对应的channel中，而这个过程就需要
    //通过fd来找到fd对应的channel，所以我们必须要搞一个fd和channel的管理器，也就是搞一个hash table来存储两者之间的映射关系
private:
    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->GetEvents();//获取Channel中fd关心的事件
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if(ret < 0) ELOG("EPOLL_CTL FAILED!!");
    }
    /* 红黑树是否已经挂上该channel(fd+event)了 */
    bool HasChannel(Channel *channel)
    {
        auto it = _chans.find(channel->Fd());
        if(it == _chans.end()) return false;
        return true;
    }
public:
    Poller() 
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS);//随便给个值就行，但给的值必须是大于0的
        if(_epfd < 0)
        {
            ELOG("EPOLL CREATE ERROR!!");
            abort();
        }
    }
    /* Channel中既包含了操作的套接字描述符，又包含了描述符上关心的事件有哪些，还包含了事件就绪时的回调函数 */
    void UpdateEvents(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if(ret == false)
        {
            _chans.insert(std::make_pair(channel->Fd(), channel));//不存在则将fd和channel的映射关系插入到哈希表中
            Update(channel, EPOLL_CTL_ADD);//不存在则将fd+care_event添加到红黑树上去
            return;
        }
        Update(channel, EPOLL_CTL_MOD);//存在则modify红黑树节点中fd所关心的事件内容
        return;
    }
    void RemoveEvents(Channel *chan)
    {
        auto it = _chans.find(chan->Fd());
        if(it != _chans.end()) { _chans.erase(it); } 
        Update(chan, EPOLL_CTL_DEL);
    }
    void Monitor(std::vector<Channel*> *active)//开始监控，返回活跃的channel数组，表示某些fd上的某些关心的事件已经ready了 
    {
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);// -1就是阻塞监控
        /* 如果一直没有事件就绪，那么epoll_wait就会阻塞监控住，服务器进程执行流会一直阻塞在这里 */
        if(nfds < 0)
        {
            if(errno == EINTR) return;//阻塞被信号打断
            else 
            {
                ELOG("EPOLL WAIT ERROR: %s\n", strerror(errno));
                abort();
            }
        }
        for(int i = 0; i < nfds; i++)
        {
            auto it = _chans.find(_evs[i].data.fd);//通过fd找到对应的channel
            assert(it != _chans.end());
            it->second->SetRevents(_evs[i].events);//设置channel事件管理器中已经ready的事件
            active->push_back(it->second);
        }
    }
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class Timer
{
private:
    uint64_t _tid;     // 定时任务对象的id
    int _timeout;      // 定时任务的超时时间
    TaskFunc _timed_task; // 定时器对象要执行的定时任务的回指指针，指向外部用户自己定义的任务函数
    ReleaseFunc _release;
    bool _canceled; // 该定时任务是否被取消
public:
    Timer(uint64_t tid, int timeout, TaskFunc task_cb) 
        : _tid(tid), _timeout(timeout), _timed_task(task_cb), _canceled(false) 
    {}
    void SetDestruct(ReleaseFunc release_cb) { _release = release_cb; }
    int DelayTime() { return _timeout; }
    void Cancel() { _canceled = true; }
    ~Timer()
    {
        if (_canceled == false) _timed_task(); 
        _release(); // 用于删除时间轮中weakptr保存的定时器对象信息
    }
};
class TimerQueue
{
private:
    int _tick; 
    int _capacity; 
    std::vector<std::vector<std::shared_ptr<Timer>>> _wheel; 
    std::unordered_map<uint64_t, std::weak_ptr<Timer>> _timers;
    EventLoop *_loop;
    int _timer_fd;//将linux系统的定时器timerfd融入到TimerQueue中，实现出一个秒级定时任务的时间轮出来
    std::unique_ptr<Channel> _timer_chan;
private:
    void ReleaseWeakptr(uint64_t tid)
    {
        auto it = _timers.find(tid);
        if (it != _timers.end()) _timers.erase(tid);
    }
    static int ActivateTimer()//激活秒级定时器
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ELOG("TIMERFD CREATE TAILED!");
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;//第一次超时时间设置为1s
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;//第一次超时之后，每隔1s就算作一次超时
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &itime, NULL);//启动定时器，让linux系统每隔1s向定时器文件描述符中写入1次超时次数的数据
        return timerfd;
    }
    int ReadTimerfd()
    {
        u_int64_t timeout;
        int ret = read(_timer_fd, &timeout, sizeof timeout);
        if(ret < 0)
        {
            ELOG("READ TIMERFD FAILED");
            abort();
        }
        return timeout;
    }
    void RunTimer()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear();//析构Timer定时器对象
    }
    void HandleTime()//当timerfd的可读事件就绪时，就会执行该回调函数，读出超时的次数，析构tick秒针沿途经历过的各个Timer定时器对象
    {
        int count = ReadTimerfd();//count表示经历了多少s
        for(int i = 0; i < count; i++) RunTimer();
    }    
private:
    void AddTimerPriv(uint64_t tid, int timeout, TaskFunc task_cb)
    {
        //1.构造一个shared_ptr管理的定时器Timer对象，同时设置一下Timer对象的成员变量，让Timer对象析构时，释放掉时间轮保存的Timer信息
        std::shared_ptr<Timer> sp(new Timer(tid, timeout, task_cb));
        sp->SetDestruct(std::bind(&TimerQueue::ReleaseWeakptr, this, tid));
        // 2.将该任务对象添加到时间轮当中
        int pos = (_tick + timeout) % _capacity;
        _wheel[pos].push_back(sp);
        _timers[tid] = std::weak_ptr<Timer>(sp);
    }
    void FlushTimerPriv(uint64_t tid)
    {
        // 1.通过保存的weak_ptr定时器对象，构造一个shared_ptr出来，将shared_ptr插入到wheel时间轮当中
        auto it = _timers.find(tid);
        if (it == _timers.end())
            return;
        std::shared_ptr<Timer> sp = it->second.lock();
        // 2.获取定时任务的延迟时间
        int delay = sp->DelayTime();
        // 3.确定添加sp的位置
        int pos = (_tick + delay) % _capacity; // tick走的下标的范围是[0, 59]
        _wheel[pos].push_back(sp);        // 刷新定时任务
    }
    void CancelTimerPriv(uint64_t tid)
    {
        // 1.通过保存的weak_ptr定时器对象，构造一个shared_ptr出来，通过shared_ptr拿到定时器对象的详细信息
        auto it = _timers.find(tid);
        if (it == _timers.end())
            return;
        std::shared_ptr<Timer> sp = it->second.lock();
        // 2.取消定时任务
        if(sp) sp->Cancel();
    }
public:
    TimerQueue(EventLoop *loop)
        : _capacity(60), _tick(0), _wheel(_capacity), _loop(loop),
        _timer_fd(ActivateTimer()), _timer_chan(new Channel(loop, _timer_fd)) 
    {
        _timer_chan->SetReadCallBack(std::bind(&TimerQueue::HandleTime, this));
        _timer_chan->CareReadEvent();
    }
    void AddTimer(uint64_t tid, int timeout, TaskFunc task_cb);
    void FlushTimer(uint64_t tid);
    void CancelTimer(uint64_t tid);
    bool HasTimer(uint64_t tid)//虽然这个接口存在线程安全问题，不能被主线程调用。但我们可以保证该接口只会在EventLoop内部线程被调用!
    {
        auto it = _timers.find(tid);
        if (it == _timers.end()) return false;
        return true;
    }
};

using Func = std::function<void()>;
class EventLoop : boost::noncopyable
{
private:
    std::thread::id _thread_id;
    int _event_fd; // create a file descriptor for event notification(具象化描述：事件通知机制，唤醒可能阻塞住的epoll_wait)
    std::unique_ptr<Channel> _efd_chan;//eventloop对象销毁时，由于智能指针管理，所以_efd_chan也会随之销毁
    Poller _poller;
    std::vector<Func> _taskqueue;//向上层提供的一个任务队列
    std::mutex _mutex; 
    TimerQueue _timer_queue;//定时销毁任务
private:
    void ExecTasks()//执行任务池中的所有任务
    {
        /* 如果给队列加锁，执行队列中的任务就需要加锁，而向队列中push任务就需要解锁，这样的效率很低。
        所以我们直接做一个vector级别的置换，将任务队列中的任务转移到另一个vector中，这样就可以保证任务队列的线程安全。 */
        std::vector<Func> tmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _taskqueue.swap(tmp);
        }
        for(auto &f : tmp) f();
    }
    int CreateEventfd()
    {
        int init = 0;//eventfd计数器的初始值
        int efd = eventfd(init, EFD_CLOEXEC | EFD_NONBLOCK);//对efd的操作都是非阻塞的
        if (efd < 0)
        {
            ELOG("CREATE EVENTFD FAILED!!");
            abort();
        }
        return efd;
    }
    void ReadEventfd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof res);
        if(ret <= 0)
        {
            if(errno == EINTR || errno == EAGAIN) return;//信号打断 + efd对应的内核接收缓冲区中没有数据
            ELOG("READ EVENTFD FAILED!!");
            abort();
        }
    }
    void WakeupEpollwait()
    {
        uint64_t val = 1;//必须是8字节的数据
        int ret = write(_event_fd, &val, sizeof val); 
        if(ret <= 0)
        {
            if(errno == EINTR) return;//信号打断
            ELOG("READ EVENTFD FAILED!!");
            abort();
        }
    }
public:
    EventLoop()
        :_thread_id(std::this_thread::get_id()), _event_fd(CreateEventfd()), _efd_chan(new Channel(this, _event_fd)), 
        _timer_queue(this)
    {
        _efd_chan->SetReadCallBack(std::bind(&EventLoop::ReadEventfd, this));
        _efd_chan->CareReadEvent();//都是静态操作，将关心的事件挂到红黑树上，只有_poller的monitor开始监控后才算是动态操作
    }
    void AssertInLoop() { assert(_thread_id == std::this_thread::get_id()); }
    bool IsLoopThread() { return _thread_id == std::this_thread::get_id(); }
    void UpdateEvents(Channel *chan) { _poller.UpdateEvents(chan); }
    void RemoveEvents(Channel *chan) { _poller.RemoveEvents(chan); }
    void AddTimer(uint64_t id, int timeout, const TaskFunc &cb) { _timer_queue.AddTimer(id, timeout, cb); }
    void FlushTimer(uint64_t id) { _timer_queue.FlushTimer(id); }
    void CancelTimer(uint64_t id) { _timer_queue.CancelTimer(id); }
    bool HasTimer(uint64_t id) { return _timer_queue.HasTimer(id); }
    void RunInLoop(const Func &cb)
    {
        /* 判断将要执行的任务是否处于当前线程中，如果是，则直接执行，不是则压入任务池，等待就绪事件被处理完毕之后统一执行 */
        if(IsLoopThread()) cb();
        else PushInTasks(cb);
    }
    void PushInTasks(const Func &cb) //将不是当前线程执行的任务，压入到任务池
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _taskqueue.push_back(cb);
        }
        WakeupEpollwait();//压入任务池后，立马唤醒epoll_wait，让LoopThread线程继续向下进行就绪事件处理+执行任务池中的任务
    }
    void RunLoop()
    {
        while (true)
        {
            // 1.事件监控
            std::vector<Channel *> actives;
            _poller.Monitor(&actives);//epoll_wait红黑树上所有的fd，将所有ready的fd所对应的channel输出到参数actives中
            // 2.就绪事件处理
            for (auto &chan : actives) chan->HandleReadyEvent();//连接超时销毁的任务放在就绪事件处理中完成，timerfd
            // 3.执行任务池中的任务
            ExecTasks();
        }
    }
};

class LoopThread
{
private:
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop *_loop;   //注意这里定义的是一个指针，而不是直接定义对象，因为EventLoop对象的实例化必须放到线程的入口函数里面进行
    std::thread _thread; 
private:
    void ThreadEntry()
    {
        //1.实例化EventLoop对象(我们在栈上创建对象，而不是堆，其目的就是将EventLoop对象的生命周期和LoopThread的生命周期进行绑定)
        //  EventLoop被实例化出来的那一刻，其内部的成员变量_thread_id就被初始化好了，因为执行EventLoop的构造函数的线程是LoopThread线程
        EventLoop loop;
        //2.唤醒有可能阻塞在_cond上的线程
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        //3.运行EventLoop模块的RunLoop
        loop.RunLoop();
    }
public:
    LoopThread() 
        : _loop(NULL), _thread(std::thread(&LoopThread::ThreadEntry, this)) // 创建线程，设定线程入口函数 
    {}
    EventLoop* GetEventLoop()
    {
        EventLoop *loop = NULL;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&](){ return _loop != NULL; });//_loop为空就阻塞等待，wait阻塞等待时，就会释放互斥锁
            loop = _loop;
            /* 实现线程间的同步操作:
            避免线程创建了，但此时EventLoop对象还没被实例化出来，此时如果直接调用GetEventLoop，则会返回一个NULL，
            所以我们实现LoopThread和mainThread的同步操作，mainThread调用GetEventLoop后，返回的一定是有效的EventLoop对象指针 */
            /* 线程同步的原理: 
            通过锁的阻塞行为来在某一特定条件下阻塞一个线程，从而让另一个线程在完成某件功能后唤醒阻塞的线程，以此来实现线程间协调工作。 */
        }
        return loop;
    }
};

class LoopThreadPool
{
private:
    int _thread_count;
    int _next_loop_idx;//下一个eventloop的索引
    EventLoop *_baseloop; //若从属线程数量为0，则事件的监控，就绪事件的处理，执行任务池中的任务等所有操作都在主线程的_baseloop内完成
    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _eventloops;
public:
    LoopThreadPool(EventLoop *baseloop) :_thread_count(0), _next_loop_idx(0), _baseloop(baseloop) {}
    void SetThreadCount(int count) { _thread_count = count; }
    void CreateAllLoopThread()
    {
        if(_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _eventloops.resize(_thread_count);
            for(int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _eventloops[i] = _threads[i]->GetEventLoop();
            }
        }
    }
    EventLoop * GetNextEventLoop()
    {
        // 1.若线程数量为0，则每次返回的都是baseloop。事件监控，就绪事件，任务队列等操作都在baseloop内完成
        if(_thread_count == 0) return _baseloop;
        // 2.若线程数量大于0，则每次采用RR轮转的思想，从_eventloops中循环遍历取出eventloop分配给Connection连接
        _next_loop_idx = (_next_loop_idx + 1) % _thread_count;
        return _eventloops[_next_loop_idx];
    }
};

class Any//保存任意自定义类型的应用层协议，应用层协议最终还是面向对象的，像该Http服务器中的HttpReqContext类型，还有可能是其他的自定义类型
{
private:
    //设计思想：在Any类里面搞一个多态，派生类用于保存任意类型的数据，而Any用于保存基类对象的指针，父类指针正好可以指向子类对象
    class holder// (holder -> 持有人 placeholder -> 占位符)
    {
    public:
        holder() {}
        virtual ~holder() {}
        virtual const std::type_info &type() = 0; // 纯虚函数，迫使子类进行重写
        virtual holder *clone() = 0;
        // 针对一个子类对象克隆出新的对象出来，因为未来可能会进行Any对象的拷贝构造，所以我们也要能够让存储数据的对象也拷贝出来一份
    };
    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}
        virtual const std::type_info &type() { return typeid(T); }
        virtual holder *clone() { return new placeholder<T>(_val); }
        virtual ~placeholder() {}
        // 默认析构，对于内置类型不处理，对于自定义类型，调用他们类内的析构函数
        // 能执行析构函数进行资源的清理，则先执行析构函数。然后就调用operator delete完成对象所占用空间的释放销毁
    public:
        T _val; // 保存任意类型的数据
    };
private:
    holder *_data_ptr;
    /*holder不是模板类，所以holder定义指针变量_data_ptr时，无须带有特定的类型，因为人家不是模板类！
    而需要是模板类来负责保存任意类型数据对象的是placeholder这个类，这样就可以完成Any的最主要的两个功能*/
    // 1.保存任意类型的数据(placeholder模板类来负责保存) 2.构造Any对象时，无须带有特定的类型(->Any的成员变量不带有模板参数)
public:
    Any() : _data_ptr(NULL) {}
    //模板函数，调用时，类型可根据参数进行推导
    template <class T>
    Any(const T &val) : _data_ptr(new placeholder<T>(val)) {}
    Any(const Any &other)
        : _data_ptr(other._data_ptr ? other._data_ptr->clone() : NULL)
    // 如果other中的父类指针不为空，则克隆一份新的数据对象出来，并将对象的地址拷贝给当前Any类对象保存的父类指针中，
    // 如果other中的父类指针为空，则也将当前Any对象中保存的父类指针置为NULL
    {}
    ~Any()
    {
        delete _data_ptr; //调用_data_ptr指向的placeholder对象内的析构函数，清理资源后调用free，释放_data_ptr变量所占用的空间
        _data_ptr = NULL;
    }
    template <class T>
    T *get() // 返回Any保存的数据对象的指针
    {
        // 用户想要获取的数据类型和Any中保存的数据对象的类型必须一致
        assert(typeid(T) == _data_ptr->type());
        return &((placeholder<T> *)_data_ptr)->_val;
    }
    Any &swap(Any &other)
    {
        std::swap(_data_ptr, other._data_ptr);
        return *this;
    }
    template <class T>
    Any &operator=(const T &val)
    {
        Any(val).swap(*this);//将构造出来的临时Any对象与当前的Any对象进行交换，则临时对象销毁时，原来的Any对象所保存的数据也就释放了
        return *this;
    }
    Any &operator=(const Any &other)
    {
        Any(other).swap(*this);
        return *this;
    }
};

class Connection;
using ConnPtr = std::shared_ptr<Connection>;
typedef enum {INCOMPLETE, ESTABLISHED, TOBECLOSED, CLOSE} ConnState;
//连接刚建立成功(各项回调函数还没被设置好) --> 通信状态(各项回调函数都被设置好了) --> 
//连接待关闭状态(等到处理完连接Buffer中滞留的数据后，状态转为关闭状态) --> 关闭状态
/* 如果用户要关闭连接，应该直接进行close吗？当然不可以，因为连接的收发缓冲区中可能还有数据滞留没被处理呢！
所以我们需要一个连接的状态来判断当前连接能否直接close，如果还处于通信状态，那说明还有滞留数据没被处理，则应该等到数据处理完毕之后
再进行连接的close */
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _conn_id;
    int _sockfd;
    bool _inact_flag;
    EventLoop *_loop; //连接所关联的EventLoop (connection : Channel : TimerTask : EventLoop : LoopThread = 1 : 1 : 1 : 1 : 1)
    ConnState _state;
    Socket _socket;
    Channel _chan;
    Buffer _in_buffer;
    Buffer _out_buffer;
    Any _context;
private:
    /* 下面这四个回调函数，是由TcpServer模块来设置的，而TcpServer内的回调其实是使用者来设置的 */
    using EstablishedCallBack = std::function<void(const ConnPtr&)>;
    using MessageCallBack = std::function<void(const ConnPtr&, Buffer *)>;//收到消息后的业务逻辑处理函数，用户设置
    using CloseCallBack = std::function<void(const ConnPtr&)>;
    using AnyEventCallBack = std::function<void(const ConnPtr&)>;
    EstablishedCallBack _established_cb_usr;
    MessageCallBack _message_cb_usr;
    CloseCallBack _close_cb_usr;
    AnyEventCallBack _anyevent_cb_usr;
    CloseCallBack _close_cb_svr;
    /* 组件内部设置的连接关闭回调，TcpServer模块内会把所有的连接管理起来，所以ConnPtr的最后一份引用技术会保存到TcpServer内的哈希表管理器中 */
private:
    void HandleRead()
    {
        //1.接收sockfd的数据放到Buffer里
        char buf[65536];
        ssize_t ret = _socket.NonBlockRecv(buf, 65535);
        if (ret < 0) 
        { 
            return ShutdownPriv();// -1表示的是连接断开，即使出错了也不能直接关闭连接，需要看一下收发缓冲区中有没有数据滞留
        }
        _in_buffer.WriteAndMoveIdx(buf, ret);
        //2.调用_message_cb进行业务逻辑处理
        if(_in_buffer.ReadableSize() > 0)
        {
            /* shared_from_this从当前对象自身，获取自身的shared_ptr管理对象，当前对象所在类必须继承一个模板类enable_shared_from_this
            模板类内部会保存一份当前对象的weak_ptr智能指针，shared_from_this()接口其实就是通过weak_ptr来获取的shared_ptr对象 */
            return _message_cb_usr(shared_from_this(), &_in_buffer);
        }
    }
    void HandleWrite()
    {
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadAddress(), _out_buffer.ReadableSize());
        if(ret < 0)//返回值<0，代表send出错，连接应该是断开了，则release连接
        {
            if(_in_buffer.ReadableSize() > 0) _message_cb_usr(shared_from_this(), &_in_buffer);
            return Release();
            /* 将实际的释放连接操作压入到任务池中进行，因为在HandleReadyEvent中处理IO事件后还会刷新连接的活跃度，
            一旦HandleWrite函数中把连接真正释放了，那么刷新连接活跃度就会导致内存访问错误，所以我们把连接的释放操作，放到就绪事件处理之后进行
            也就是压入到任务池中进行 */
        }
        _out_buffer.MoveReadIdx(ret);//千万不要忘了将读偏移位置向后偏移
        if(_out_buffer.ReadableSize() == 0)
        {
            /* 关闭写事件监控，因为写缓冲区中没有数据了，没啥可写的，自然就不需要关心sockfd上的写事件了 
            否则会出现一种情况，sockfd上的写事件不断被触发，epoll_wait不断检测到写事件就绪，但_out_buffer中却一直没有数据发送，
            这其实就是在浪费epoll_wait的资源，所以在_out_buffer中没有数据之后，我们关闭其对于写事件的监控*/
            _chan.CancelWriteEvent();
            /* 如果当前连接的状态是待关闭的状态，那在发送完_out_buffer中的内容之后，就应该释放连接，同时将连接状态置为关闭状态 */
            if(_state == TOBECLOSED) return Release();
        }
    }
    void HandleClose()
    {
        /* 一旦连接挂断了，套接字就什么都干不了了，这个挂断一般是客户端主动挂断连接，此时服务器即使_out_buffer中有数据，那也什么都做不了
        因为连接已经断开了。如果此时_in_buffer中还有数据的话，那我们服务器就捎带的进行一下数据的业务逻辑处理，然后再实际释放掉连接！ */
        if(_in_buffer.ReadableSize() > 0) _message_cb_usr(shared_from_this(), &_in_buffer);
        return Release();
    }
    void HandleError() { return HandleClose(); }
    void HandleAnyEvent()
    {
        /* sockfd上触发任意事件的回调: 1.刷新连接的活跃度，延迟定时销毁任务的执行  2.调用组件使用者的任意事件回调进行处理 */
        if(_inact_flag == true) { _loop->FlushTimer(_conn_id); }
        if(_anyevent_cb_usr) { _anyevent_cb_usr(shared_from_this()); }
    }
private:
    void RefineConnectPriv()
    {
        //1.修改连接状态     2.启动读事件监控    3.执行用户的回调函数
        assert(_state == INCOMPLETE);
        _state = ESTABLISHED;
        _chan.CareReadEvent();
        if(_established_cb_usr) _established_cb_usr(shared_from_this());
        /* 为什么不能在Connection的构造函数中开启读事件监控呢？
        对于聊天室使用场景来讲，我们希望在连接刚建立完成后，给聊天室内的online user发送xxx上线的消息。如果在构造函数中开启读事件监控，
        则可能导致我们还没来得及调用_established_cb_usr给在线用户发送xxx上线的消息时，连接上已经收到了大量的消息了，我们希望的是，先
        发送xxx上线的消息，然后再进行连接上的通信。
        所以理论上放到构造函数中也是可以的，只不过这有点不符合回调函数在连接建立后回调的规则，我们希望的是在连接正式建立完成之后再去
        进行_established_cb_usr的回调，所以我们将开启读事件的监控放到RefineConnectPriv中进行，开启读事件监控之后，
        立马进行_established_cb_usr的回调，这样就会先给在线用户广播xxx上线消息，然后再开始连接上的通信！ */
    }
    void ReleasePriv()//实际释放连接的接口
    {
        //1.修改连接状态，改为实际释放状态
        _state = CLOSE;
        //2.移除连接的事件监控
        _chan.RemoveEvents();
        //3.close sockfd
        _socket.Close();
        //4.如果定时任务队列中还有定时销毁任务，则应该取消定时销毁任务的执行
        if(_loop->HasTimer(_conn_id))
        {
            CloseInactPriv();
        }
        //5.调用组件使用者设置的关闭回调函数
        /* 避免先移除服务器内部保存的实际连接信息而导致Connection被释放，再去调用组件使用者的回调可能会出错！所以我们先调用组件使用者的cb
        然后再调用组件内部的关闭连接回调，进行连接管理信息的移除 */
        if(_close_cb_usr) _close_cb_usr(shared_from_this());//调用组件使用者设置好的连接关闭时的回调
        if(_close_cb_svr) _close_cb_svr(shared_from_this());//移除TcpServer内部所管理的连接信息
    } 
    void SendPriv(const Buffer &buf)
    {
        /* 这个接口并不是真正的send数据，而是将要发送的数据放到_out_buffer里面，然后启动可写事件监控，等可写事件就绪后，就会回调HandleWrite
        在HandleWrite内部才会进行数据的实际发送。换句话说，触发可写事件之后，我们才进行数据的发送 */
        if(_state == CLOSE) return;
        _out_buffer.WriteAndMoveIdx(buf);
        if(_chan.IsWritable() == false)
        {
            //如果当前连接可写事件没有就绪，那就关心可写事件，等待可写事件就绪之后，自动回调HandleWrite进行数据的发送
            _chan.CareWriteEvent();
        }
    }
    void ShutdownPriv()/* 这个关闭操作并非实际的连接释放操作，还需要判断_in_buffer有没有数据待处理，_out_buffer待发送 */
    {
        _state = TOBECLOSED;//设置连接为即将关闭状态
        if(_in_buffer.ReadableSize() > 0) 
        /* 有可能_in_buffer中的数据不完整，上层一直不取出_in_buffer中的数据，导致_in_buffer中一直有数据存在 。
           接下来，要么就是HandleWrite发送数据出错(因为对方已经关闭连接)导致释放连接，要么就是没有要发送的数据直接释放连接，要么就是HandleWrite发送完数据
           发送连接标志位是TOBECLOSED，那么在发送完数据之后就会关闭连接。（大概实际关闭连接是以上三种情况）*/
        {
            if(_message_cb_usr) _message_cb_usr(shared_from_this(), &_in_buffer);
        }
        if(_out_buffer.ReadableSize() > 0)
        {
            if(_chan.IsWritable() == false) _chan.CareWriteEvent();
        }
        else if(_out_buffer.ReadableSize() == 0)// 如果发送缓冲区中的数据已经发送完了，那我们就直接关闭释放连接
        {
            ReleasePriv();
        }
    }
    void OpenInactPriv(int seconds)
    {
        //1.将标志位置为true
        _inact_flag = true;
        //2.如果当前定时销毁任务已经存在，那就刷新一下即可。如果不存在，我们才会添加
        if(_loop->HasTimer(_conn_id)) { return _loop->FlushTimer(_conn_id); }
        _loop->AddTimer(_conn_id, seconds, std::bind(&Connection::Release, this));
        /* 不能直接释放连接，直接释放可能会导致内存访问错误，原因见client4.cpp测试过程
        必须将释放连接的操作压入任务池，在就绪事件处理完成后，再去释放连接 */
    }
    void CloseInactPriv()
    {
        _inact_flag = false;
        if(_loop->HasTimer(_conn_id)) _loop->CancelTimer(_conn_id);
    }
    void SwitchProtocolPriv(const Any& context, const EstablishedCallBack &esta_cb, const MessageCallBack& mess_cb,
                         const CloseCallBack& close_cb, const AnyEventCallBack& anyevent_cb)
    { 
        _context = context, _established_cb_usr = esta_cb, _message_cb_usr = mess_cb, 
        _close_cb_usr = close_cb, _anyevent_cb_usr = anyevent_cb; 
    }
public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        :_conn_id(conn_id), _sockfd(sockfd), _inact_flag(false), _loop(loop), _state(INCOMPLETE), _socket(_sockfd)
        , _chan(loop, sockfd)
    {
        _chan.SetCloseCallBack(std::bind(&Connection::HandleClose, this));
        _chan.SetAnyEventCallBack(std::bind(&Connection::HandleAnyEvent, this));
        _chan.SetReadCallBack(std::bind(&Connection::HandleRead, this));
        _chan.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));
        _chan.SetErrorCallBack(std::bind(&Connection::HandleError, this));
        /* 注意: chan的读事件监控不要在构造函数里面进行. (原因见->RefineConnectPriv)*/
    }
    ~Connection(){ DLOG("RELEASE CONNECTION: %p", this); }
    int Fd() { return _sockfd; }
    uint64_t Id() { return _conn_id; }
    bool IsEstablished() { return _state == ESTABLISHED; }
    void SetContext(const Any& context) { _context = context; }
    Any* GetContext() { return &_context; }
    void SetEstablishedCbUsr(const EstablishedCallBack& cb) { _established_cb_usr = cb; }
    void SetMessageCbUsr(const MessageCallBack& cb) { _message_cb_usr = cb; }
    void SetCloseCbUsr(const CloseCallBack& cb) { _close_cb_usr = cb; }
    void SetCloseCbSvr(const CloseCallBack& cb) { _close_cb_svr = cb; }    
    void SetAnyEventCbUsr(const AnyEventCallBack& cb) { _anyevent_cb_usr = cb; }
    void RefineConnect() { _loop->RunInLoop(std::bind(&Connection::RefineConnectPriv, this)); }
    void Send(const char *data, size_t len)
    {
        //连接上接收到消息后，用户业务逻辑处理之后，可能要发送数据，但发送数据的实际操作只能在Eventloop线程内完成
        //因为光一个连接内部所关联的channel在多线程环境下，都有可能产生线程安全，所以向外提供的这些函数最终都得在EventLoop的任务队列中执行
        Buffer buf;
        buf.WriteAndMoveIdx(data, len);//data指向的空间可能会被释放掉，所以我们这里做一个数据的拷贝，拷贝到buf里面
        _loop->RunInLoop(std::bind(&Connection::SendPriv, this, std::move(buf)));
        //RunInLoop只要一执行，就会唤醒阻塞的epoll_wait，让eventloop继续向下处理就绪事件+执行任务池的任务
    }
    void Shutdown() { _loop->RunInLoop(std::bind(&Connection::ShutdownPriv, this)); }
    void Release() { _loop->PushInTasks(std::bind(&Connection::ReleasePriv, this)); }
    //执行Release的线程一定是EventLoop线程自己，所以直接调用PushInTasks就行，当然你想要调用RunInLoop也可以，就是多执行了一个函数而已
    void OpenInactiveRelease(int sec) { _loop->RunInLoop(std::bind(&Connection::OpenInactPriv, this, sec)); }
    void CloseInactiveRelease() { _loop->RunInLoop(std::bind(&Connection::CloseInactPriv, this)); }
    void SwitchProtocol(const Any& context, const EstablishedCallBack &esta_cb, const MessageCallBack& mess_cb,
                         const CloseCallBack& close_cb, const AnyEventCallBack& anyevent_cb)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::SwitchProtocolPriv, this, context, esta_cb, mess_cb, close_cb, anyevent_cb));
    }
    /* 切换协议(切换数据的处理方式): 重置上下文和阶段性回调处理函数---这个接口必须在EventLoop的线程中立即执行，不可以压入到任务池执行
    存在一种情况，上层使用者调用该接口进行协议升级，但协议升级的这个API是放到任务池中进行的，很有可能这个API还没执行呢，但此时连接上新就绪
    了事件，那么此时数据的处理还是会按照原来的协议方式进行处理，这样肯定是不行的！！！
    所以这个接口必须是EventLoop自己的线程来执行，外部的线程不能调用该接口，我们认为外部直接调用协议切换函数的这种方式是错误的！
    （目前该接口在服务器组件中没有用到） */
};

class Accepter
{
private:
    using ReadCallBack = std::function<void(int)>;
    Socket _lissock; 
    EventLoop *_loop;//用于对监听套接字进行事件监控 
    Channel _chan; 
    ReadCallBack _read_cb;
private:
    void HandleRead()//监听套接字的读事件回调函数: 获取新连接 + 为新连接分配一个Connection对象
    {
        int newfd = _lissock.Accept();
        if(newfd < 0) return;
        if(_read_cb) _read_cb(newfd);
    }
    int CreateServerConn(uint16_t port)
    {
        bool ret = _lissock.CreateServerConn(port);
        assert(ret == true);
        return _lissock.Fd();
    }
public:
    Accepter(EventLoop *loop, uint16_t port) 
        :_lissock(CreateServerConn(port)), _loop(loop), _chan(loop, _lissock.Fd())
    { 
        _chan.SetReadCallBack(std::bind(&Accepter::HandleRead, this)); 
    }
    void SetReadCallBack(const ReadCallBack& cb) { _read_cb = cb; }
    void CareReadEvent() { _chan.CareReadEvent(); }
};

class TcpServer
{
private:
    uint64_t _conn_id;
    int _timeout; 
    int _port;      //服务器监听的端口
    bool _inact_flag;
    EventLoop _baseloop;
    /* 
        1.主线程创建该baseloop，该baseloop关联的线程id就是主线程的id，满足能够实现单Reactor服务器的demand
        其实这个baseloop就是主Reactor，负责获取可能产生的高并发连接请求，而其余的eventloop就是从属Reactor，负责通信连接上监控事件的处理，
        两者都是事件驱动处理模式
        2.如果eventloop数量为0，则baseloop自己负责lissock和sockfd两种套接字的三件套处理Monitor -> HandleReadyEvent -> ExecTasks
        baseloop自己调用RunLoop()，不断进行三件套的循环处理
        3.如果eventloop数量大于0，则baseloop负责lissock的三件套处理，eventloop负责sockfd的三件套处理，将两种套接字分到主从Reactor中
        进行事件监控，就绪事件处理，执行任务池任务
    */
    Accepter _accepter;
    LoopThreadPool _pool; 
    std::unordered_map<uint64_t, ConnPtr> _conns; 
private:
    using EstablishedCallBack = std::function<void(const ConnPtr&)>;
    using MessageCallBack = std::function<void(const ConnPtr&, Buffer *)>;//收到消息后的业务逻辑处理函数，用户设置
    using CloseCallBack = std::function<void(const ConnPtr&)>;
    using AnyEventCallBack = std::function<void(const ConnPtr&)>;
    EstablishedCallBack _established_cb_usr;
    MessageCallBack _message_cb_usr;
    CloseCallBack _close_cb_usr;
    AnyEventCallBack _anyevent_cb_usr;
private:
    void AddTimerTaskPriv(const Func &task, int sec)
    {
        _conn_id++;//这里把连接的id拿过来借用给定时任务
        _baseloop.AddTimer(_conn_id, sec, task);//用户自己所添加的定时任务是在baseloop里面执行的
    }
    void ReleaseConnPtrPriv(ConnPtr connptr)
    {
        int id = connptr->Id(); 
        auto it = _conns.find(id);
        if(it != _conns.end()) _conns.erase(it);//销毁最后一份哈希表中存储的conn连接对象，实现连接的真正释放
    }
    void ReleaseConnPtr(ConnPtr connptr) 
    { 
        /* 服务器模块的通信连接的彻底销毁，最终其实是在baseloop里面执行的，其他从属线程调用的_close_cb_svr，其实是将释放连接的操作，
        压入到了baseloop的任务队列中执行的，压入到任务池，并不是就泡在里面迟迟不执行了，相反很快就会执行，因为只要压入任务池，就会立马
        唤醒可能阻塞的epoll_wait，向下进行就绪事件处理，然后执行任务池的任务 */
        _baseloop.RunInLoop(std::bind(&TcpServer::ReleaseConnPtrPriv, this, connptr)); 
    }
    void NewConnection(int sockfd)
    {
        _conn_id++;
        ConnPtr conn(new Connection(_pool.GetNextEventLoop(), _conn_id, sockfd));//Connection的构造函数会设置好sockfd的五个事件回调函数
        conn->SetEstablishedCbUsr(_established_cb_usr);
        conn->SetMessageCbUsr(_message_cb_usr);
        conn->SetCloseCbUsr(_close_cb_usr);
        conn->SetAnyEventCbUsr(_anyevent_cb_usr);
        conn->SetCloseCbSvr(std::bind(&TcpServer::ReleaseConnPtr, this, std::placeholders::_1));
        if(_inact_flag == true) conn->OpenInactiveRelease(_timeout);
        conn->RefineConnect();//开启通信套接字sockfd的读事件监控
        /* <定时任务的添加, 开启读事件监控> 这两个操作之间有时序问题，必须先进行定时任务的添加，然后再开启读事件的监控
         ---反过来错误的原因: 
         先开启读事件监控时，可能立马有事件到来，这时候anyeventCallback会进行定时任务的刷新，但这个时候时间轮里面还没有定时任务呢，
         因为是后进行的定时任务添加，所以反过来是不可行的！
         ---不会产生时序问题的原因：
         开启定时任务销毁的这个函数会被压入到任务池中，而PushInTasks的操作会唤醒epoll_wait，主线程运行完PushInTasks后，回到1241行
         继续向下开启读事件监控，但在主线程运行完PushInTasks的一瞬间，EventLoop线程就会HandleReadyEvent了，下一步就执行任务池中的任务，
         而当主线程开启读事件监控时，EventLoop线程就算是最慢最慢，他也会进行到HandleReadyEvent，而处理的事件中一定是不包含当前连接的，
         所以一定不会刷新当前连接相关的定时任务，因为EventLoop线程早就epoll_wait结束了，即使当前我们的连接上有读事件就绪，那也一定是
         EventLoop线程执行完任务池中的任务，而任务中就包含了添加当前连接对应的定时销毁任务，所以当前连接读事件就绪时，定时任务早就被
         添加到时间轮里面了，此时刷新定时任务就不会出错了。*/        
        _conns.insert(std::make_pair(_conn_id, conn));
    }
public:
    TcpServer(int port) 
        :_port(port), _conn_id(0), _inact_flag(false), _accepter(&_baseloop, port), _pool(&_baseloop)
    {
        _accepter.SetReadCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _accepter.CareReadEvent();//都是静态操作
    }
    void Run()
    {
        /*服务器模块的运行，具象化描述就是 事件驱动处理模式 == eventloop.runloop(); 三件套:Monitor -> HandleReadyEvent -> ExecTasks
        baseloop和eventloop都是最终调用RunLoop运行，不断执行三件套*/
        _pool.CreateAllLoopThread();
        _baseloop.RunLoop();
    }
    void SetThreadCount(int count) { _pool.SetThreadCount(count); }
    void SetEstablishedCallBack(const EstablishedCallBack& cb) { _established_cb_usr = cb; }
    void SetMessageCallBack(const MessageCallBack& cb) { _message_cb_usr = cb; }
    void SetCloseCallBack(const CloseCallBack& cb) { _close_cb_usr = cb; }
    void SetAnyEventCallBack(const AnyEventCallBack& cb) { _anyevent_cb_usr = cb; }
    void OpenInactiveRelease(int timeout) { _timeout = timeout; _inact_flag = true; }
    //用于添加用户自身的定时任务的接口
    void AddTimerTask(const Func &task, int sec) { _baseloop.RunInLoop(std::bind(&TcpServer::AddTimerTaskPriv, this, task, sec)); }
};

void Channel::RemoveEvents() { _loop->RemoveEvents(this); }
void Channel::UpdateEvents() { _loop->UpdateEvents(this); }
/* 对定时任务的操作只能放到loopthread线程中运行，不能让主线程运行，因为loopthread内部在不断loop循环运行的过程中，是会对_timers这个
哈希表进行操作的，例如释放时间轮中某一index的Timer对象，而添加，刷新，取消定时器的操作也会涉及到_timers的操作，所以为了保证_timers的
线程安全，我们将主线程想要对时间轮进行定时任务的添加，刷新，取消等操作，并不直接执行，而是间接放到EventLoop内部的任务队列中执行！*/
void TimerQueue::AddTimer(uint64_t tid, int timeout, TaskFunc task_cb) { _loop->RunInLoop(std::bind(&TimerQueue::AddTimerPriv, this, tid, timeout, task_cb)); }
void TimerQueue::FlushTimer(uint64_t tid) { _loop->RunInLoop(std::bind(&TimerQueue::FlushTimerPriv, this, tid)); }
void TimerQueue::CancelTimer(uint64_t tid) { _loop->RunInLoop(std::bind(&TimerQueue::CancelTimerPriv, this, tid)); }

class NetWork
{
public:
    NetWork()
    {
        // DLOG("SIGPIPE IGNORE INIT!");
        signal(SIGPIPE, SIG_IGN);
        /* 当连接断开后，如果服务器还继续向连接发送数据，此时就会触发异常SIGPIPE，操作系统就会给我们的服务器进程发送SIGPIPE，
        进程就会终止退出，所以我们让服务器忽略掉SIGPIPE信号 */
    }
};
static NetWork nw;
#endif